package com.atguigu.gmallrealtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmallrealtime.bean.TableProcess;
import com.atguigu.gmallrealtime.common.Constant;
import com.atguigu.gmallrealtime.util.HBaseUtil;
import com.atguigu.gmallrealtime.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @author yhm
 * @create 2023-09-23 14:28
 */
public class DimSinkFunc extends RichSinkFunction<Tuple2<TableProcess, JSONObject>> {
    StatefulRedisConnection<String, String> redisConnection ;
    @Override
    public void open(Configuration parameters) throws Exception {
        redisConnection = RedisUtil.getAsyncRedisConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeAsyncRedisConnection(redisConnection);
    }

    @Override
    public void invoke(Tuple2<TableProcess, JSONObject> value, Context context) throws Exception {
        // 1. 获取要写出的数据
        TableProcess tableProcess = value.f0;
        JSONObject jsonObject = value.f1;
        // 2. 判断写出的类型
        String opType = jsonObject.getString("op_type");
        jsonObject.remove("op_type");
        String rowKey = jsonObject.getString(tableProcess.getSinkRowKey());
        // 3. 根据不同的类型写出数据
        if ("delete".equals(opType)){
            // 删除hbase中对应的数据
            // namespace tableName rowKey 列族 列名
            HBaseUtil.dropColumn(Constant.HBASE_NAMESPACE,tableProcess.getSinkTable(),rowKey,tableProcess.getSinkFamily(),jsonObject.keySet());
            // 同时将redis中的数据删除
            redisConnection.async().del(tableProcess.getSinkTable() + ":" +rowKey);

        }else if ("insert".equals(opType) || "update".equals(opType) || "bootstrap-insert".equals(opType)){
            // 覆盖写入对应的数据
            HBaseUtil.putCell(Constant.HBASE_NAMESPACE,tableProcess.getSinkTable(),rowKey,tableProcess.getSinkFamily(),jsonObject);
            if ( "update".equals(opType)){
                redisConnection.async().del(tableProcess.getSinkTable() + ":" +rowKey);
            }

        }else {
            // 错误的数据类型
        }
    }


}
